package com.hfzy.ihk.common.core.elasticsearch.base.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.hfzy.ihk.common.core.elasticsearch.ElasticsearchLocalCache;
import com.hfzy.ihk.common.core.elasticsearch.ElasticsearchOperation;
import com.hfzy.ihk.common.core.elasticsearch.base.OperationHandler;
import com.hfzy.ihk.common.core.elasticsearch.base.OperationManager;
import com.hfzy.ihk.common.core.elasticsearch.entity.OperationErrorMsg;
import com.hfzy.ihk.common.core.elasticsearch.repository.common.OperationErrorMsgRepository;
import com.hfzy.ihk.common.facade.BaseEntity;
import com.hfzy.ihk.common.util.string.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.DependsOn;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by Administrator on 2018/5/28.
 * 基础job，每个服务都需运行
 * @author wws
 */
@Component
@DependsOn({"operationErrorMsgRepository"})
@ConditionalOnProperty(prefix = "linkcall.elasticsearch",name = "enable")
public class CommonRetryOperationMsgJobs {

    private static final Logger logger = LoggerFactory.getLogger(CommonRetryOperationMsgJobs.class);

    @Autowired
    OperationErrorMsgRepository operationErrorMsgRepository;

    @PostConstruct
    public void init() throws Exception {
        logger.info("init CommonRetryOperationMsgJobs bean");
    }

    @Scheduled(cron = "0 0/1 * * * ?")
    public void reTryOptErrorMsg(){
        logger.debug("-----retry reTryOptErrorMsg-------");
       Page<OperationErrorMsg> page = operationErrorMsgRepository.findAll(PageRequest.of(0,1000));

       logger.debug("-----retry reTryOptErrorMsg-------:{}",page.getTotalElements());
        List<OperationErrorMsg> rmList = new ArrayList<>();
       page.getContent().stream().forEach(transactionalErrorMsg->{

           transactionalErrorMsg.setRetryTimes(transactionalErrorMsg.getRetryTimes()+1);
           if(todoOpteration(transactionalErrorMsg)) {
               rmList.add(transactionalErrorMsg);
           }

       });
        rmList.forEach(msg-> operationErrorMsgRepository.delete(msg));
        logger.debug("-----retry reTryOptErrorMsg-------over,size:{}",page.getTotalElements());
    }

    @Scheduled(cron = "0/10 * * * * ?")
    public void reTryCacheOptErrorMsg(){

        logger.debug("-----retry reTryCacheOptErrorMsg-------");

        if(ElasticsearchLocalCache.getOptErrorMsgMemoryCache().size()<=0) {
            return;
        }

        List<OperationErrorMsg> rmList = new ArrayList<>();
        ElasticsearchLocalCache.getOptErrorMsgMemoryCache().forEach(transactionalErrorMsg->{

            logger.debug("-----retry reTryCacheOptErrorMsg-------"+ JSON.toJSONString(transactionalErrorMsg));
            if(todoOpteration(transactionalErrorMsg)) {
                rmList.add(transactionalErrorMsg);
            }

        });
        rmList.forEach(msg-> ElasticsearchLocalCache.remove(msg));
        logger.debug("-----retry reTryCacheOptErrorMsg---over----");

    }

    protected boolean todoOpteration(OperationErrorMsg msg){

        String key = StringUtil.strConnect(msg.getOptTable(),":",msg.getOptMethod());

        OperationHandler handler = OperationManager.getHandlerMap().get(key);

        if(handler==null){
            return false;
        }

        ElasticsearchOperation operation = new ElasticsearchOperation(msg.getTxId(),msg.getOptTable(),msg.getOptMethod(),
               JSONArray.parseArray(msg.getJsonArrayMessage()));

        /**标识来自job重试*/
        operation.setFromJobRun(true);

        return handler.handle(operation);
    }

}
